package com.study.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

/**
 * @author
 * @date 2022年09月10日 21:06
 * @description
 */
public class CustomConsumerSeekTime {


    public static void main(String[] args) {


        // 配置

        Properties properties = new Properties();


        // 连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.160:9092");


        // 反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 指定group id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test2");



        // 创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        // 指定主题
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");

        // 订阅主题
        kafkaConsumer.subscribe(topics);


        // 指定位置消费
        Set<TopicPartition> assignment = kafkaConsumer.assignment();


        while (assignment.size() == 0) {
            kafkaConsumer.poll(Duration.ofSeconds(1));
            assignment = kafkaConsumer.assignment();
        }

        // 把时间转换为对应的offset

        HashMap<TopicPartition, Long> topicPartitionLongHashMap = new HashMap<>();


        for (TopicPartition topicPartition : assignment) {
            topicPartitionLongHashMap.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 60 * 60 * 100);
        }
        Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = kafkaConsumer.offsetsForTimes(topicPartitionLongHashMap);


        for (TopicPartition topicPartition : assignment) {
            OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsetAndTimestampMap.get(topicPartition);

            // 此处是由于有3个分区无数据是后面创建的，所以加个判空
            if (offsetAndTimestamp == null) {
                continue;
            }
            kafkaConsumer.seek(topicPartition, offsetAndTimestamp.offset());
        }

        // 消费数据
        while (true) {
            for (ConsumerRecord<String, String> consumerRecord : kafkaConsumer.poll(Duration.ofSeconds(1))) {
                System.out.println(consumerRecord);
            }
        }


    }
}
